package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.CommentBean;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.text.DecimalFormat;
import java.time.Duration;

public class DwsInteractionComment {
    public static void main(String[] args) throws Exception {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO　2 设置状态后端
                 /*
                 env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
                 env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
                 env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
                 env.setStateBackend(new HashMapStateBackend());
                 env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
                 System.setProperty("HADOOP_USER_NAME", "atguigu");
                  */
        //TODO 3 读取dwd_interaction_cource_comment主题数据
        String topicName = "dwd_interaction_cource_comment";
        String groupId = "dws_interaction_comment";
        DataStreamSource<String> commentStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

        //TODO 4 转化格式
        SingleOutputStreamOperator<CommentBean> beanStream = commentStream.map(new MapFunction<String, CommentBean>() {
            @Override
            public CommentBean map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                return CommentBean.builder()
                        .userCount(1L)
                        .courseId(jsonObject.getJSONObject("data").getString("course_id"))
                        .userCount(Long.parseLong("1"))
                        .avgScore(jsonObject.getJSONObject("data").getDouble("review_stars"))
                        .ts(jsonObject.getLong("ts"))
                        .build();
            }
        });
        //TODO 5  聚合开窗
        SingleOutputStreamOperator<CommentBean> beanWithWatermarkStream =
                beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<CommentBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<CommentBean>() {
            @Override
            public long extractTimestamp(CommentBean commentBean, long l) {
                return commentBean.getTs() * 1000L;
            }
        }));

        SingleOutputStreamOperator<CommentBean> applyStream = beanWithWatermarkStream.keyBy(new KeySelector<CommentBean,
                String>() {
            @Override
            public String getKey(CommentBean commentBean) throws Exception {
                return commentBean.getCourseId();
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10))).apply(new WindowFunction<CommentBean, CommentBean,
                String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<CommentBean> iterable,
                              Collector<CommentBean> collector) throws Exception {
                //累计评价人数
                long userCount = 0L;

                //累计评价分数
                Double sumScore = 0.0;

                //5星好评人数
                long starCount = 0L;

                for (CommentBean commentBean : iterable) {
                    userCount += commentBean.getUserCount();
                    sumScore += commentBean.getAvgScore();
                    if (commentBean.getAvgScore() == 5.0) {
                        starCount++;
                    }
                }

                //保留两位小数
                DecimalFormat df = new DecimalFormat("#.00");

                //5星好评率
                String fivestarrating = df.format((double)starCount / (double) userCount) ;


                //平均评分
                String avgScore = df.format(sumScore / userCount);

                CommentBean commentBean = iterable.iterator().next();
                commentBean.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                commentBean.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                commentBean.setAvgScore(Double.parseDouble(avgScore));
                commentBean.setGoodComments(Double.parseDouble(fivestarrating));
                commentBean.setUserCount(userCount);
                commentBean.setTs(System.currentTimeMillis());
                collector.collect(commentBean);

            }
        });


        //TODO 6 写入clickHouse中
        applyStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_interaction_comment values" +
                "(?,?,?,?,?,?,?)"));


        //TODO 7 执行任务
        env.execute(groupId);


    }
}
